package openjdk.server;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * Created by boot on 1/12/16.
 * 使用NIO实现网络编程
 * 1. AcceptJob 负责接收新的请求
 * ...
 * <p>
 * 2. ProcessJob 负责具体的请求处理
 * 通过 clientSelector.selectNow() 能够得到当前是否有客户端就绪
 * 而clientSelector.selectKeys().iterator() 能够对当前已就绪的客户
 * 端进行迭代。只有当客户端就绪时，如SelectionKey.isReadable() 为true时
 * ProcessJob 才会进行相应的I/O处理.
 * Selector 机制使得线程不必等待客户端的就绪，当客户羰端还没有就绪时,可以处理
 * 其它的请求，提高了服务的端的并发吞吐能力，降低了资源消耗。
 */
public class NIOServer {

    static class AcceptJob implements Runnable {
        private boolean keepRunning = true;
        private Selector clientSelector;
        private Selector serverSelector;

        public AcceptJob(Selector clientSelector, Selector serverSelector) {
            this.clientSelector = clientSelector;
            this.serverSelector = serverSelector;
        }

        @Override
        public void run() {
            try {
                while (keepRunning) {
                    int n = serverSelector.selectNow();
                    if (n <= 0) continue;
                    Iterator<SelectionKey> it = serverSelector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        // 处理业务逻辑
                        SelectionKey key = it.next();
                        if (key.isAcceptable()) {
                            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            // 与Selector一起使用时，Channel必须处于非阻塞模式下
                            // 这意味着不能将FileChannel与Selector一起使用，因为FileChannel不能切换到非阻塞模式,而套接字通道都可以。
                            socketChannel.configureBlocking(false); // Note
                            socketChannel.register(clientSelector, SelectionKey.OP_READ);
                        }
                        // 业务逻辑处理完成后删除当前的SelectionKey
                        it.remove();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        public void stopRunning() {
            this.keepRunning = false;
        }
    }


    static class ProcessJob implements Runnable {
        private boolean keepRunning = true;
        private Selector clientSelector;

        public ProcessJob(Selector clientSelector) {
            this.clientSelector = clientSelector;
        }

        @Override
        public void run() {
            int cnt = 0;
            ByteBuffer buf = ByteBuffer.allocate(1024);
            try {
                while (keepRunning) {
                    int n = clientSelector.selectNow();
                    if (n <= 0) continue;
                    Iterator<SelectionKey> it = clientSelector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        if (key.isReadable()) {
                            // client socket channel.
                            SocketChannel socketChannel = (SocketChannel) key.channel();

                            while ((cnt = (socketChannel.read(buf))) > 0) {
                                System.out.write(buf.array(), 0, cnt);

                                // 也可以使用如下两行代码代替 System.out.write(buf.array(),0,cnt);
                                //buf.flip();
                                //System.out.println(Charset.forName(System.getProperty("file.encoding")).decode(buf));

                                // 将ByteBuffer 清空,不是真的清空噢
                                buf.clear();
                                socketChannel.write(buf); // 将收到的数据再写回去
                            }

                            socketChannel.close();
                        }
                    }
                    it.remove();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        public void stopRunning() {
            this.keepRunning = false;
        }
    }


    public static void main(String[] args) throws Exception {
        Selector clientSelector = Selector.open();
        Selector serverSelector = Selector.open();

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(serverSelector, SelectionKey.OP_ACCEPT);

        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(9999));


        final AcceptJob acceptJob = new AcceptJob(clientSelector, serverSelector);
        final ProcessJob processJob = new ProcessJob(clientSelector);

        Thread acceptThread = new Thread(acceptJob);
        Thread processThread = new Thread(processJob);

        acceptThread.start();
        processThread.start();


        // 50秒后自动停止服务器，让接收线程和处理线程结束run()方法，
        //  main 线程自动退出。
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(50 * 1000);
                    acceptJob.stopRunning();
                    processJob.stopRunning();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}
